package com.bwie.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.io.file.FileNameUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.bwie.domain.Class;
import com.bwie.domain.Shop;
import com.bwie.domain.Student;
import com.bwie.domain.Type;
import com.bwie.es.EsStudent;
import com.bwie.mapper.ClassMapper;
import com.bwie.mapper.ShopMapper;
import com.bwie.mapper.StuMapper;
import com.bwie.mapper.TypeMapper;
import com.bwie.service.StuService;
import com.bwie.utils.R;
import com.bwie.vo.PageInfoVo;
import com.bwie.vo.ShopVo;
import com.github.tobato.fastdfs.domain.StorePath;
import com.github.tobato.fastdfs.service.FastFileStorageClient;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * @Author: lrz
 * @Date: 2024/11/4 09:03
 * @Description:
 **/
@Service
public class StuServiceImpl implements StuService {

    @Autowired
    StuMapper stuMapper;

    @Autowired
    ClassMapper classMapper;

    @Autowired
    TypeMapper typeMapper;

    @Autowired
    ShopMapper shopMapper;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    FastFileStorageClient fastFileStorageClient;

    @Override
    public R list(PageInfoVo pageInfoVo) {

        Integer pageNum=0;
        if(pageInfoVo.getPageNum()>0){
            pageNum = pageInfoVo.getPageNum()-1;
        }
        PageRequest pageRequest = PageRequest.of(pageNum, pageInfoVo.getPageSize());

        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        if(StringUtils.isNotEmpty(pageInfoVo.getKeyWord())){
            boolQueryBuilder.should(QueryBuilders.matchQuery("classType", pageInfoVo.getKeyWord()));
            boolQueryBuilder.should(QueryBuilders.matchQuery("stuName", pageInfoVo.getKeyWord()));
        }
        if(pageInfoVo.getTypeId()!=null){
            boolQueryBuilder.must(QueryBuilders.termQuery("flagId", pageInfoVo.getTypeId()));
        }

        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        nativeSearchQueryBuilder.withPageable(pageRequest).withQuery(boolQueryBuilder);

        SearchHits<EsStudent> search = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), EsStudent.class);

        long totalHits = search.getTotalHits();
        ArrayList<EsStudent> records = new ArrayList<>();
        for (SearchHit<EsStudent> searchHit : search.getSearchHits()) {
            EsStudent content = searchHit.getContent();

            records.add(content);
        }

        HashMap<String, Object> hashMap = new HashMap<>();
        hashMap.put("totals",totalHits);
        hashMap.put("records",records);

        return R.OK(hashMap);
    }

    @Override
    public R addStudent(Student student) {
        QueryWrapper<Student> studentQueryWrapper = new QueryWrapper<>();
        studentQueryWrapper.eq("stu_name",student.getStuName());
        Student student1 = stuMapper.selectOne(studentQueryWrapper);
        if(student1!=null){
            return R.ERROR(500,"该学生已存在");
        }

        QueryWrapper<Type> typeQueryWrapper = new QueryWrapper<>();
        typeQueryWrapper.eq("id",student.getXk());
        Type type = typeMapper.selectOne(typeQueryWrapper);

        student.setClassType(type.getName());


        student.setHgfs(90.00);
        student.setFlagId(0);

        stuMapper.insert(student);

        QueryWrapper<Class> classQueryWrapper = new QueryWrapper<>();
        classQueryWrapper.eq("id",student.getBjId());
        Class aClass = classMapper.selectOne(classQueryWrapper);
        aClass.setCount(aClass.getCount()+1);


        classMapper.updateById(aClass);

        kafkaTemplate.send("test_topic", JSON.toJSONString(student));

        return R.OK();
    }

    @Override
    public R classList() {
        List<Class> classes = classMapper.selectList(null);
        return R.OK(classes);
    }

    @Override
    public R updateStudent(Student student) {

        if(student.getFlagId()==2){
            return R.ERROR(500,"请更改你真实的成绩");
        }

        //获取学号后四位
        String xh = student.getXh().substring(student.getXh().length()-4);
        if(!xh.equals(student.getHsw())){
            return R.ERROR(500,"身份证后四位不正确，请重新输入");
        }

        stuMapper.updateById(student);

        return R.OK();
    }

    @Override
    public R listClass(PageInfoVo pageInfoVo) {
        Page<Class> classPage = new Page<>(pageInfoVo.getPageNum(), pageInfoVo.getPageSize());
        QueryWrapper<Class> classQueryWrapper = new QueryWrapper<>();
        classQueryWrapper.eq("id",pageInfoVo.getTypeId());
        classQueryWrapper.orderByDesc("sbl");
        Page<Class> classPage1 = classMapper.selectPage(classPage, classQueryWrapper);
        return R.OK(classPage1);
    }

    @Override
    public R shmsg(MultipartFile file) throws IOException {
        String originalFilename = file.getOriginalFilename();
        String extName = FileNameUtil.extName(originalFilename);

        StorePath storePath = fastFileStorageClient.uploadFile(file.getInputStream(), file.getSize(), extName, null);
        String url = "http://192.168.80.192:8888/"+storePath.getFullPath();
        return R.OK(url);
    }

    @Override
    public R plDelete(ArrayList<Integer> ids) {
        stuMapper.deleteBatchIds(ids);
        return R.OK();
    }

    @Override
    public R shopList() {
        ArrayList<ShopVo> shopVos =  shopMapper.shopList();

        ArrayList<String> xlist = new ArrayList<>();
        ArrayList<Double> ylist = new ArrayList<>();
        for (ShopVo shopVo : shopVos) {
            xlist.add(shopVo.getXname());
            ylist.add(shopVo.getYvalue());
        }

        HashMap<String, Object> hashMap = new HashMap<>();
        hashMap.put("xlist",xlist);
        hashMap.put("ylist",ylist);

        return R.OK(hashMap);
    }

    @KafkaListener(topics = "test_topic")
    public void testKafka(String message, Acknowledgment acknowledgment){
        System.out.println("接受到消息了："+message);

        acknowledgment.acknowledge();

        System.out.println("接收完成了！！！");
    }

}
